package com.sqoop.client.client;

import com.sqoop.commons.receive.MsgListener;
import com.sqoop.core.MsgFutrue.MsgFutrue;
import com.sqoop.core.futruecache.FutureCache;
import com.sqoop.core.futruecache.MsgFutureCache;
import com.sqoop.core.msg.Msg;
import com.sqoop.core.msg.MsgDestination;
import com.sqoop.core.request.Response;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.concurrent.Future;

/**
 * Created by zw on 2017/11/13.
 */
public class MsgReceiver {

    private static Logger logger = LogManager.getLogger(MsgReceiver.class.getName());
    private MsgListener msgListener = null;
    private FutureCache futureCache = new MsgFutureCache();

    public void register(MsgDestination msgDestination){
        msgListener.register(msgDestination);
    }

    public class ReceiverRunnable implements Runnable{

        @Override
        public void run() {
            logger.info("MsgReceiver start!!!");
            while (true) {
                Msg msg = msgListener.getMsg();
                if (msg == null) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    continue;
                }
                Response response = (Response) msg;
                logger.info("receive a response ,id is {}" + response.getMsgid());
                String msgid = msg.getMsgid();
                Future future = futureCache.get(msgid);
                ((MsgFutrue) future).done(response);
            }
        }
    }

    public MsgReceiver(MsgListener msgListener, FutureCache futureCache) {
        this.msgListener = msgListener;
        this.futureCache = futureCache;
        Thread t = new Thread(new ReceiverRunnable());
        t.start();
    }


}